<?php

/**
 * 消费者
 * */

namespace App\Lib;

use AMQPConnection;
use AMQPChannel;
use AMQPExchange;
use AMQPQueue;
use App\Http\Model\Hint;
use App\Http\Services\RoomMsgServices;

class ConsumerRoomChampionUpdate
{
    const QUEUE_ROUTE = 'global.room.champion.update';// 队列路由键

    protected $queueConfig = [];// 队列总配置
    protected $queueRoute = [];// 队列路由
    protected $connectConfig = [];// 连接配置

    protected $consumerMode = 'callback';// 默认阻塞回调

    public function __construct()
    {
        $this->queueConfig = config('queue') ?? [];

        $this->connectConfig = [
            'host' => $this->queueConfig['RABBITMQ_HOST'] ?? '127.0.0.1',
            'port' => $this->queueConfig['RABBITMQ_PORT'] ?? 5672,
            'vhost' => $this->queueConfig['RABBITMQ_VHOST'] ?? '/',
            'login' => $this->queueConfig['RABBITMQ_LOGIN'] ?? 'guest',
            'password' => $this->queueConfig['RABBITMQ_PASSWORD'] ?? 'guest',
            'prefetch_count' => 20,
            'heartbeat' => 30,
        ];
    }

    public function consumer()
    {
        $code = Hint::SUCCESS;
        $msg = Hint::getErrCodeMsg($code);
        $data = [];

        do {
            try {
                // 队列配置项 路由配置 交换类型
                if (empty($this->queueConfig) or empty($this->queueConfig['queueRoute'])) {
                    $code = Hint::ERR_FAIL;
                    $msg = '队列配置项初始' . Hint::getErrCodeMsg($code);
                    break;
                }

                if (!array_key_exists(self::QUEUE_ROUTE, $this->queueConfig['queueRoute'])) {
                    $code = Hint::ERR_PARAMS_CHECK_FAIL;
                    $msg = '队列路由:' . Hint::getErrCodeMsg($code);
                    break;
                }

                // TODO 队列路由
                $this->queueRoute = $this->queueConfig['queueRoute'][self::QUEUE_ROUTE];

                /*if (!extension_loaded('amqp')) {
                    $code = Hint::ERR_FAIL;
                    $msg = 'No extension amqp loaded failed !';
                    break;
                }*/

                $conn = new AMQPConnection($this->connectConfig);
                $conn->connect();
                if (!$conn->isConnected()) {
                    $code = Hint::ERR_FAIL;
                    $msg = 'rabbit-mq Connection failed !';
                    break;
                }

                $channel = new AMQPChannel($conn);
                if (!$channel->isConnected()) {
                    $code = Hint::ERR_FAIL;
                    $msg = 'rabbit-mq Connection through channel failed !';
                    break;
                }

                $exchange = new AMQPExchange($channel);
                $exchange->setFlags(AMQP_PASSIVE);//声明一个已存在的交换器的，如果不存在将抛出异常，这个一般用在consume端
                $exchange->setName($this->queueRoute['exchangeName'] ?: '');
                $exchange->setType(AMQP_EX_TYPE_TOPIC); // topic 交换类型
                $exchange->declareExchange();

                //$channel->startTransaction();

                $queue = new AMQPQueue($channel);
                $queue->setName($this->queueRoute['queueName'] ?: '');
                $queue->setFlags(AMQP_DURABLE);
                $queue->declareQueue();

                //绑定
                foreach ($this->queueRoute['routeKey'] as $value) {
                    $queue->bind($this->queueRoute['exchangeName'], $value);
                }

                if ($this->consumerMode == 'callback') {
                    $queue->consume('App\Lib\ConsumerRoomChampionUpdate::callback');
                } else {
                    write_log('consumer', 'message', []);
                    $this->message($queue);
                }

                $conn->disconnect();

            } catch (\Exception $e) {
                $code = Hint::ERR_SYS_FAIL;
                $msg = Hint::getErrCodeMsg($code) . ':' . $e->getMessage();
                break;
            }

        } while (false);

        return compact('code', 'msg', 'data');
    }

    /**
     * @todo 废弃
     * 回调处理数据
     *
     * 第一种消费方式,阻塞模式接收消息,程序一直会卡在此处
     * */
    public static function callback($envelope, $queue)
    {
        if ($envelope) {
            // sleep(1);  //sleep 1秒

            $body = $envelope->getBody();
            $body = json_decode($body, true);

            // todo log
            write_log('ConsumerRoomChampionUpdate', 'callback', $body);

            // 此处处理业务逻辑
            /*$roomMsgService = new RoomMsgServices();
            $roomMsgService->roomChampionUpdate($body['room_id'],$body['user_id'] ,$body['val'] ,$body['type']);*/
            // 逻辑处理成功 应答消费掉此消息
            $queue->ack($envelope->getDeliveryTag());

        } else {
            echo 'No message' . PHP_EOL;
        }
    }

    /**
     * 第二种消费方式,非阻塞
     * */
    public function message($queue)
    {
        $message = $queue->get();
        if (!empty($message)) {
            $body = $message->getBody();

            // todo log
            write_log('ConsumerRoomChampionUpdate', 'message', json_decode($body, true));

            // 此处处理业务逻辑

            $queue->ack($message->getDeliveryTag());    //应答，代表该消息已经消费
        }
    }
}

